/**
 * Created by dev-001 on 2017/3/6.
 * procucer和consumer暂时放在一起
 */
var open = require('amqplib').connect('amqp://localhost');

// Publisher
export async function producer(q, msg){
    open.then(function(conn) {
        return conn.createChannel();
    }).then(function(ch) {
        return ch.assertQueue(q).then(function(ok) {
            console.log(ok)
            let _msg = typeof msg == 'object' ? JSON.stringify(msg) : msg;
            return ch.sendToQueue(q, new Buffer(_msg));
        });
    }).catch(console.warn);
}


export async function consumer(q, callback) {
    // Consumer
    open.then(function(conn) {
        return conn.createChannel();
    }).then(function(ch) {
        return ch.assertQueue(q).then(function(ok) {
            return ch.consume(q, async function(msg) {
                if (msg !== null) {
                    let content = msg.content.toString();
                    await callback(JSON.parse(content));
                    ch.ack(msg);
                }
            });
        });
    }).catch(console.warn);
}